package com.dmp.userLabel.bolt;

import java.util.Map;

import org.slf4j.Logger;

import com.dmp.common.string.StringUtils;
import com.dmp.storm.bolt.BaseMapBolt;
import com.dmp.userLabel.factory.CarLaberFactory;
import com.dmp.userLabel.factory.ConcreteFactory;
import com.dmp.userLabel.factory.Factory;
import com.dmp.userLabel.factory.ICarLaber;
import com.dmp.userLabel.factory.ILabelRedisDao;

public class CarExtendBolt extends BaseMapBolt {
	private static final long serialVersionUID = 1L;
	private static final Logger logger = org.slf4j.LoggerFactory.getLogger(CarExtendBolt.class);

	private String redisPath;
	private ILabelRedisDao iLabelRedisDao;
	private String extendField;

	@Override
	public void prepare() {
		// 创建一个具体工厂
		Factory factory = new ConcreteFactory();
		iLabelRedisDao = factory.createLabelRedisDao(this.redisPath);
	}

	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		long beginTime = System.currentTimeMillis();
		Map<String, Object> _result = null;

		if (map != null) {
			_result = this.extendFields(map);
		}

		// end time
		long endTime = System.currentTimeMillis();
		float excTime = (float) (endTime - beginTime);
		if (_result != null) {
			logger.info("{schedule_out_put:[time-consuming info]AdSaveToKafkaBolt:beginTime:{},endTime:{},excTime:{}}", beginTime, endTime, excTime);
			logger.info("{schedule_out_put:CarExtendBolt:---->send bolt:{}}", _result);
		}
		return _result;
	}

	private Map<String, Object> extendFields(Map<String, Object> map) {
		// 这里写业务逻辑
		boolean flag = false;
		if (StringUtils.isNotBlank(extendField)) {
			String[] extendFieldArray = extendField.split(",");

            /* ！！！多个特性标签，生成多个处理标签的类，逻辑上有点问题 ！！！*/
			for (String extendField : extendFieldArray) {
				ICarLaber userLaber = CarLaberFactory.createCarLaber(extendField);
				if (userLaber != null) {
					flag = userLaber.dealCarLabel(map, flag, iLabelRedisDao);
				} else {
					logger.error("{carLaber:{},do not find ICarLaber,please check this type!}", extendField);
				}
			}
		}
		if (!flag) {
			return null;
		}
		return map;
	}

	public String getRedisPath() {
		return redisPath;
	}

	public void setRedisPath(String redisPath) {
		this.redisPath = redisPath;
	}

	public String getExtendField() {
		return extendField;
	}

	public void setExtendField(String extendField) {
		this.extendField = extendField;
	}

}
